In [1]:
from bson import SON
import pymongo
from pymongo import MongoClient, GEOSPHERE
from datetime import datetime
import pandas as pd
import folium
from folium import plugins
import cufflinks as cf
cf.set_config_file(world_readable=True,offline=True)

def get_session(db_name):
    """
    :param db_name: Nombre de la base de datos de los incidentes y los distritos
    :return: Objeto de tipo DataBase con una conexión a la base de datos local
    """
    client = MongoClient()
    db = client[db_name]
    return db


def create_indexes(db):
    """
    GEOSPHERE: para procesar coordinadas esféricas
    :param db: Añade un índice sobre los campos que contienen información geoespacial
    """
    db.incidents.create_index([("Location", GEOSPHERE)])
    db.neighbours.create_index([("the_geom", GEOSPHERE)])
    db.incidents.create_index([("Date", pymongo.ASCENDING)])


def first_query(db):
    """
    :param db: referencia a la sesión de la bd
    :return: Esta función obtiene los incidentes que están a una distancia máximo de 1000 metros
    desde un punto representado por coordinadas geográficas en formato geojson
    """
    # Montamos la query
    query_incidents = {
        "Location" :{
            "$near": {
                "$geometry" : SON([
                    ("type", "Point"),
                    ("coordinates", [-122.42158168136999, 37.7617007179518])
                ]),
                "$maxDistance": 1000
            }
        }
    }
    # Ejecutamos la querry sobre la colección de incidencias
    query_results = db.incidents.find(query_incidents)
    df = pd.DataFrame(list(query_results))
    return df


def second_query(db):
    """
    :param db: referencia a la sesión de la bd
    :return: devuelve el distrito que contiene las coordinadas usadas en el
    operado de intersección
    """
    # Montamos la query
    query_distrito = {"the_geom":
        {"$geoIntersects":
            {"$geometry": SON([
                ("type", "Point"),
                ("coordinates", [-122.42158168136999, 37.7617007179518])])
            }
        }
    }
    # Ejecutamos la querry sobre la colección de incidencias
    query_results = db.neighbours.find_one(query_distrito)
    return query_results


def third_query(db):
    """
    :param db: referencia a la sesión de la bd
    :return: Devuelve el todos los incidentes, en dataframe, de un distrito, usando
    operadores geo-espaciales, la consulta se realiza en fases sobre las dos
    colecciones, primero encontramos el distrito, y luego buscamos todos los
    incidentes que tienen las coordinadas dentro del polígono
    """
    # Empezamos con el distrito
    query_distrito = {"the_geom": {"$geoIntersects": {
        "$geometry": SON([("type", "Point"), ("coordinates", [-122.42158168136999, 37.7617007179518])])}}}
    distrito = db.neighbours.find_one(query_distrito)
    # Ahora encontramos los incidentes
    query_incidents = {"Location": {"$geoWithin": {"$geometry": distrito['the_geom']}}}
    query_results = db.incidents.find(query_incidents)
    df = pd.DataFrame(list(query_results))
    return df


def since_february(db):
    """
    :param db: referencia a la sesión de la bd
    :return: Devuelve los incidentes que han ocurrido desde febrero de 2018
    """
    fecha = datetime(2018, 2, 1)
    incidents = db.incidents.find({"Date": {"$gt": fecha}})
    df = pd.DataFrame(list(incidents))
    return df


def date_querry(op, sdate, edate):
    """
    :param opr: operador B: between dates, L: less than, G: greater than
    :return: devuelve el filtro de la consulta según lo que se recibe por parámetro
    en op
    """
    switcher = {
        'B': {"Date": {"$gte": sdate, "$lte": edate}},  # Between
        'GE': {"Date": {"$gte": sdate}},              # Greater than or equal
        'LE': {"Date": {"$lte": sdate}},              # less than or equal
    }
    return switcher.get(op)


def generic_date_search(db, op, sdate, *args, **kwargs):
    """
    :param db: referencia a la sesión de bd
    :param op: operador para seleccionar
    :param sdate: primera fecha, mandaterio
    :param args:
    :param kwargs: Contiene argumento opcional de la seguna fecha
    :return: incidentes que cumplen con el filtro sobre fechas
    """
    edate = kwargs.get('edate', None)
    if not edate is None:
        query = date_querry(op, sdate, edate)
    else:
        query = date_querry(op, sdate, None)
    query_results = db.incidents.find(query)
    df = pd.DataFrame(list(query_results))
    return df


def draw_map(ds):
    """
    Esta función recibe un conjunto de incidentes y los dibuja en un mapa usando el paquete folium,
    el mapa se guarda en un fichero.
    :param ds: dataset de incidentes en formato de DataFrame
    """
    incid_map = folium.Map(location=[37.7617007179518, -122.42158168136999], zoom_start=11, tiles='Stamen Terrain')
    marker_cluster = plugins.MarkerCluster().add_to(incid_map)
    for name, row in ds.iterrows():
        folium.Marker([row["Y"], row["X"]], popup=row["Descript"]).add_to(marker_cluster)
    incid_map.save('incidents.html')
    return incid_map


def draw_heatmap(df):
    """
    Esta función recibe un conjunto de incidentes y los dibuja en un mapa de calor que
    se guarda en un fichero html
    :param ds: dataset de incidentes en formato de DataFrame
    """
    heat_map = folium.Map(location=[37.7617007179518, -122.42158168136999], zoom_start=11, tiles='Stamen Terrain')
    heat_map.add_child(plugins.HeatMap([[row["Y"], row["X"]] for name, row in df.iterrows()]))
    heat_map.save('heat_map_incidets.html')
    return heat_map


def total_per_category(db):
    """
    Esta función usa el framework de aggregate para obtener el total de cada categoría
    :param db: referencia a la sesión de bd
    :return: Dataframe composed of two columns, categories and count of each category
    """
    pipeline = [
        {"$group": {"_id": "$Category", "count": {"$sum": 1}}},
        {"$sort": SON([("count", -1), ("_id", -1)])}
    ]
    aggregate_results = db.incidents.aggregate(pipeline)
    return pd.DataFrame(list(aggregate_results))
In [2]:
sfdb = get_session("san_francisco_incidents")  # Creamos una sesión
create_indexes(sfdb)  # Nos aseguramos de que estén los índices
In [3]:
fq = first_query(sfdb)  # Consulta geoespacial con operador $near
fq.head()
Out[3]:
Address Category Date DayOfWeek Descript IncidntNum Location PdDistrict PdId Resolution Time X Y _id
0 18TH ST / VALENCIA ST NON-CRIMINAL 2015-01-19 14:00:00 Monday LOST PROPERTY 150060275 {'coordinates': [-122.42158168136999, 37.76170... MISSION 15006027571000 NONE 14:00 -122.421582 37.761701 5ac1522e790de03ca663d972
1 VALENCIA ST / 18TH ST NON-CRIMINAL 2015-02-27 16:50:00 Friday LOST PROPERTY 150185005 {'coordinates': [-122.42158168136999, 37.76170... MISSION 15018500571000 NONE 16:50 -122.421582 37.761701 5ac1522e790de03ca664001b
2 18TH ST / VALENCIA ST NON-CRIMINAL 2014-01-16 12:00:00 Thursday LOST PROPERTY 140089047 {'coordinates': [-122.42158168136999, 37.76170... MISSION 14008904771000 NONE 12:00 -122.421582 37.761701 5ac1522e790de03ca6642fc7
3 18TH ST / VALENCIA ST LARCENY/THEFT 2014-03-12 21:30:00 Wednesday PETTY THEFT FROM A BUILDING 140212969 {'coordinates': [-122.42158168136999, 37.76170... MISSION 14021296906303 NONE 21:30 -122.421582 37.761701 5ac1522e790de03ca66440d9
4 VALENCIA ST / 18TH ST VANDALISM 2014-06-14 18:28:00 Saturday MALICIOUS MISCHIEF, VANDALISM OF VEHICLES 140494521 {'coordinates': [-122.42158168136999, 37.76170... MISSION 14049452128160 NONE 18:28 -122.421582 37.761701 5ac1522e790de03ca6646803
In [4]:
sq = second_query(sfdb)  # Consulta geoespacial con el operador $geoIntersects
In [5]:
tq = third_query(sfdb)  # Consulta geoespacial
In [6]:
# Fechas para filtro de fechas
fecha1 = datetime(2017, 12, 1)
fecha2 = datetime(2017, 12, 31)
# Buscaremos los incidentes entre dos fechas, B=Between (ver doc de la función)
date_results = generic_date_search(sfdb,'B', fecha1, edate=fecha2)
In [7]:
# Buscamos todos los incidentes de febrero
feb = since_february(sfdb)
In [8]:
# Generamos un mapa con los miles primeros incidentes
m = draw_map(feb.iloc[:1000])
hm = draw_heatmap(date_results.iloc[:1000])
In [9]:
m
Out[9]:
In [10]:
hm
Out[10]:
In [11]:
df = total_per_category(sfdb)
df
Out[11]:
_id count
0 LARCENY/THEFT 472052
1 OTHER OFFENSES 305920
2 NON-CRIMINAL 235017
3 ASSAULT 191842
4 VEHICLE THEFT 125740
5 DRUG/NARCOTIC 118702
6 VANDALISM 114324
7 WARRANTS 100316
8 BURGLARY 90232
9 SUSPICIOUS OCC 79406
10 MISSING PERSON 64134
11 ROBBERY 55205
12 FRAUD 40984
13 SECONDARY CODES 25418
14 FORGERY/COUNTERFEITING 22913
15 WEAPON LAWS 21892
16 TRESPASS 19126
17 PROSTITUTION 16664
18 STOLEN PROPERTY 11748
19 SEX OFFENSES, FORCIBLE 11507
20 DISORDERLY CONDUCT 9977
21 DRUNKENNESS 9775
22 RECOVERED VEHICLE 8716
23 DRIVING UNDER THE INFLUENCE 5616
24 KIDNAPPING 5293
25 RUNAWAY 4385
26 LIQUOR LAWS 4076
27 ARSON 3868
28 EMBEZZLEMENT 2953
29 LOITERING 2420
30 SUICIDE 1282
31 FAMILY OFFENSES 1174
32 BAD CHECKS 916
33 BRIBERY 802
34 EXTORTION 732
35 SEX OFFENSES, NON FORCIBLE 423
36 GAMBLING 342
37 PORNOGRAPHY/OBSCENE MAT 57
38 TREA 14
In [12]:
df.iplot(kind="pie", labels="_id", values="count")